package com.weizhu.swufer.core.grpc;

import cn.hutool.core.collection.CollUtil;
import com.google.gson.Gson;
import com.weizhu.swufer.constant.SwuferConstant;
import com.weizhu.swufer.core.client.RemoteChannel;
import com.weizhu.swufer.core.entity.RpcBaseInfo;
import com.weizhu.swufer.core.rpc.RpcEngine;
import io.grpc.Channel;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import org.apache.commons.pool2.BaseKeyedPooledObjectFactory;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.KeyedPooledObjectFactory;
import org.apache.commons.pool2.PooledObject;
import org.apache.commons.pool2.impl.DefaultPooledObject;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

/**
 * 本地缓存服务地址
 * 缓存服用grpc通道
 */
public class RemoteCacheManager {

    private ConcurrentHashMap<String, CopyOnWriteArrayList<String>> addressCacheMap;

    private ConcurrentHashMap<String,RemoteChannel> channelCacheMap;

    private GenericKeyedObjectPool<String, RemoteChannel> channelPool;

    private volatile static RemoteCacheManager instance;

    private RemoteCacheManager(){
        this.addressCacheMap = new ConcurrentHashMap<>();
        this.channelCacheMap = new ConcurrentHashMap<>();
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.err.println("<===== shutting down gRPC channel since JVM is shutting down =====>");
            for (RemoteChannel remoteChannel : channelCacheMap.values()) {
                ManagedChannel channel = (ManagedChannel)remoteChannel.getChannel();
                if (channel != null && !channel.isShutdown()){
                    channel.shutdown();
                }
            }
            System.err.println("<===== all grpc channel shut down =====>");
        }));
    }

    public static RemoteCacheManager getInstance(){
        if (instance == null){
            synchronized (RpcEngine.class){
                if (instance == null){
                    instance = new RemoteCacheManager();
                }
            }
        }
        return instance;
    }

    public void cacheAddress(String serviceName,String info){
        CopyOnWriteArrayList<String> infoList = addressCacheMap.get(serviceName);
        if (infoList == null){
            infoList = new CopyOnWriteArrayList<>();
            addressCacheMap.put(serviceName,infoList);
        }
        infoList.add(info);
        System.out.println("缓存了服务注册信息"+serviceName+"/"+info);
    }

    public void removeCacheAddress(String serviceName,String info){
        Gson gson = new Gson();
        RpcBaseInfo rpcBaseInfo = gson.fromJson(info, RpcBaseInfo.class);
        CopyOnWriteArrayList<String> addressList = addressCacheMap.get(serviceName);
        if (CollUtil.isEmpty(addressList)){
            return;
        }
        addressList.remove(info);
        RemoteChannel remoteChannel = channelCacheMap.get(rpcBaseInfo.getAddress());
        if (remoteChannel != null){
            ManagedChannel channel = (ManagedChannel)remoteChannel.getChannel();
            if (!channel.isShutdown()){
                channel.shutdown();
            }
            channelCacheMap.remove(rpcBaseInfo.getAddress());
        }
    }

    public List<String> getCacheAddressList(String serviceName){
        return addressCacheMap.get(serviceName);
    }


    public synchronized RemoteChannel getChannel(String addressKey){
        String[] split = addressKey.split(SwuferConstant.IP_ADDRESS_DELIMITER);
        String ip = split[0];
        int port = Integer.parseInt(split[1]);
        RemoteChannel remoteChannel;
        remoteChannel = channelCacheMap.get(addressKey);
        if (remoteChannel != null){
            return remoteChannel;
        }
        ManagedChannel channel = ManagedChannelBuilder.forAddress(ip,port)
                .usePlaintext().directExecutor().build();
        remoteChannel = new RemoteChannel(ip,port,channel);
        channelCacheMap.put(addressKey,remoteChannel);
        return remoteChannel;
    }

    public void removeChannel(String address){
        channelCacheMap.remove(address);
    }


    public void removeRemoteServiceCache(String address){
        Gson gson = new Gson();
        for (Map.Entry<String, CopyOnWriteArrayList<String>> entry : addressCacheMap.entrySet()) {
            CopyOnWriteArrayList<String> serverList = entry.getValue();
            serverList.removeIf((x)->{
                RpcBaseInfo rpcBaseInfo = gson.fromJson(x, RpcBaseInfo.class);
                return rpcBaseInfo.getAddress().equals(address);
            });
        }
    }

    public RemoteChannel getChannel(String ip,int port){
        String addressKey = ip + SwuferConstant.IP_ADDRESS_DELIMITER + port;
        return getChannel(addressKey);
    }

    public GenericKeyedObjectPool<String, RemoteChannel> getChannelPool() {
        return channelPool;
    }
}
